package com.fxd.service.impl;

import com.fxd.enums.RedisStreamEnum;
import com.fxd.service.DealNoAckMessageService;
import com.fxd.util.RedisTemplateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.List;

@Slf4j
@Service
public class DealNoAckMessageServiceImpl implements DealNoAckMessageService {

    @Autowired
    private RedisTemplateUtil redisTemplateUtil;

    @Override
    public void dealFruitsProducerStream() {
        //一次最多获取10条未ack的消息
        PendingMessages pendingMessages = redisTemplateUtil.xPendingInfo(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue(), 10);

        pendingMessages.iterator().forEachRemaining(message -> {
            //超过1小时未ack时才进行处理
            if (message.getElapsedTimeSinceLastDelivery().getSeconds() > 3600) {
                if (message.getTotalDeliveryCount() > 5) {  //消息转发5次仍然未ack，避免死信队列，主动进行消费

                } else {
                    //获取消费组下的所有消费组
                    List<String> consumerNames = redisTemplateUtil.xInfoConsumers(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue());
                    if (CollectionUtils.isEmpty(consumerNames)) {  //未找到消费者
                        return;
                    }
                    //找一个不为当前消息所属消费者的另一个消费者
                    for (String toConsumerName : consumerNames) {
                        if (!toConsumerName.equals(message.getConsumerName())) {
                            MapRecord<String, String, String> mapRecord = redisTemplateUtil.xClaim(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue(), toConsumerName, Duration.ofSeconds(30), message.getId());
                            log.info("stream: {}中consumer: {}获取到consumer: {}的消息id: {}"
                                    , mapRecord.getStream(), toConsumerName, message.getConsumerName(), mapRecord.getId());
                            log.info("消息重试获取到消息对象: {}", mapRecord);

                            //业务处理

                            //发送ack指令
                            redisTemplateUtil.xAck(mapRecord.getStream(), message.getGroupName(), mapRecord.getId());
                            break;
                        }
                    }
                }
            }
        });
    }

}
